package com.datagateway.component;

import com.datagateway.config.DataGatewayProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 配置热更新管理器
 * 支持运行时动态修改配置，无需重启应用
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
@Component
@RefreshScope
public class ConfigHotReloadManager {

    private static final Logger logger = LoggerFactory.getLogger(ConfigHotReloadManager.class);

    @Autowired
    private Environment environment;

    @Autowired
    private DataGatewayProperties dataGatewayProperties;

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    @Autowired
    private AlertManager alertManager;

    /**
     * 配置监听器注册表
     */
    private final Map<String, List<ConfigChangeListener>> configListeners = new ConcurrentHashMap<>();

    /**
     * 配置缓存
     */
    private final Map<String, Object> configCache = new ConcurrentHashMap<>();

    /**
     * 文件监听器
     */
    private WatchService watchService;

    /**
     * 定时任务执行器
     */
    private ScheduledExecutorService scheduler;

    /**
     * 配置热更新状态
     */
    private volatile boolean hotReloadEnabled = true;

    /**
     * 初始化配置热更新管理器
     */
    @PostConstruct
    public void init() {
        try {
            // 初始化文件监听器
            initFileWatcher();
            
            // 初始化定时任务
            initScheduledTasks();
            
            // 加载初始配置
            loadInitialConfigs();
            
            logger.info("配置热更新管理器初始化完成");
            
        } catch (Exception e) {
            logger.error("配置热更新管理器初始化失败", e);
            alertManager.sendSystemErrorAlert("配置热更新管理器初始化失败", e.getMessage());
        }
    }

    /**
     * 初始化文件监听器
     */
    private void initFileWatcher() {
        try {
            watchService = FileSystems.getDefault().newWatchService();
            
            // 监听配置文件目录
            String configPath = environment.getProperty("spring.config.location", "classpath:/");
            if (configPath.startsWith("classpath:")) {
                // 对于classpath配置，使用定时检查方式
                logger.info("使用定时检查方式监听classpath配置");
            } else {
                // 对于文件系统配置，使用文件监听器
                Path configDir = Paths.get(configPath);
                if (Files.exists(configDir)) {
                    configDir.register(watchService, 
                        StandardWatchEventKinds.ENTRY_MODIFY,
                        StandardWatchEventKinds.ENTRY_CREATE,
                        StandardWatchEventKinds.ENTRY_DELETE);
                    logger.info("文件监听器已启动，监听目录: {}", configDir);
                }
            }
            
        } catch (IOException e) {
            logger.error("初始化文件监听器失败", e);
        }
    }

    /**
     * 初始化定时任务
     */
    private void initScheduledTasks() {
        scheduler = Executors.newScheduledThreadPool(2);
        
        // 定时检查配置变更
        scheduler.scheduleWithFixedDelay(this::checkConfigChanges, 30, 30, TimeUnit.SECONDS);
        
        // 定时处理文件监听事件
        scheduler.scheduleWithFixedDelay(this::processFileWatchEvents, 10, 10, TimeUnit.SECONDS);
    }

    /**
     * 加载初始配置
     */
    private void loadInitialConfigs() {
        try {
            // 加载Kafka配置
            loadKafkaConfigs();
            
            // 加载Hive配置
            loadHiveConfigs();
            
            // 加载Groovy配置
            loadGroovyConfigs();
            
            // 加载告警配置
            loadAlertConfigs();
            
            logger.info("初始配置加载完成");
            
        } catch (Exception e) {
            logger.error("加载初始配置失败", e);
        }
    }

    /**
     * 加载Kafka配置
     */
    private void loadKafkaConfigs() {
        Map<String, Object> kafkaConfigs = new HashMap<>();
        kafkaConfigs.put("kafka.bootstrap-servers", environment.getProperty("kafka.bootstrap-servers"));
        kafkaConfigs.put("kafka.consumer.group-id", environment.getProperty("kafka.consumer.group-id"));
        kafkaConfigs.put("kafka.consumer.auto-offset-reset", environment.getProperty("kafka.consumer.auto-offset-reset"));
        kafkaConfigs.put("kafka.consumer.max-poll-records", environment.getProperty("kafka.consumer.max-poll-records"));
        
        configCache.putAll(kafkaConfigs);
    }

    /**
     * 加载Hive配置
     */
    private void loadHiveConfigs() {
        Map<String, Object> hiveConfigs = new HashMap<>();
        hiveConfigs.put("hive.jdbc.url", environment.getProperty("hive.jdbc.url"));
        hiveConfigs.put("hive.jdbc.username", environment.getProperty("hive.jdbc.username"));
        hiveConfigs.put("hive.jdbc.password", environment.getProperty("hive.jdbc.password"));
        hiveConfigs.put("hive.batch.size", environment.getProperty("hive.batch.size"));
        hiveConfigs.put("hive.batch.timeout", environment.getProperty("hive.batch.timeout"));
        
        configCache.putAll(hiveConfigs);
    }

    /**
     * 加载Groovy配置
     */
    private void loadGroovyConfigs() {
        Map<String, Object> groovyConfigs = new HashMap<>();
        groovyConfigs.put("groovy.script.path", environment.getProperty("groovy.script.path"));
        groovyConfigs.put("groovy.script.cache.enabled", environment.getProperty("groovy.script.cache.enabled"));
        groovyConfigs.put("groovy.script.cache.size", environment.getProperty("groovy.script.cache.size"));
        
        configCache.putAll(groovyConfigs);
    }

    /**
     * 加载告警配置
     */
    private void loadAlertConfigs() {
        Map<String, Object> alertConfigs = new HashMap<>();
        alertConfigs.put("alert.mail.enabled", environment.getProperty("alert.mail.enabled"));
        alertConfigs.put("alert.mail.smtp.host", environment.getProperty("alert.mail.smtp.host"));
        alertConfigs.put("alert.mail.smtp.port", environment.getProperty("alert.mail.smtp.port"));
        alertConfigs.put("alert.mail.recipients", environment.getProperty("alert.mail.recipients"));
        
        configCache.putAll(alertConfigs);
    }

    /**
     * 检查配置变更
     */
    private void checkConfigChanges() {
        if (!hotReloadEnabled) {
            return;
        }
        
        try {
            // 检查Kafka配置变更
            checkKafkaConfigChanges();
            
            // 检查Hive配置变更
            checkHiveConfigChanges();
            
            // 检查Groovy配置变更
            checkGroovyConfigChanges();
            
            // 检查告警配置变更
            checkAlertConfigChanges();
            
        } catch (Exception e) {
            logger.error("检查配置变更失败", e);
        }
    }

    /**
     * 检查Kafka配置变更
     */
    private void checkKafkaConfigChanges() {
        String[] kafkaKeys = {
            "kafka.bootstrap-servers",
            "kafka.consumer.group-id",
            "kafka.consumer.auto-offset-reset",
            "kafka.consumer.max-poll-records"
        };
        
        for (String key : kafkaKeys) {
            String newValue = environment.getProperty(key);
            String oldValue = (String) configCache.get(key);
            
            if (!Objects.equals(newValue, oldValue)) {
                logger.info("检测到Kafka配置变更: {} = {} -> {}", key, oldValue, newValue);
                configCache.put(key, newValue);
                notifyConfigChange(key, oldValue, newValue);
            }
        }
    }

    /**
     * 检查Hive配置变更
     */
    private void checkHiveConfigChanges() {
        String[] hiveKeys = {
            "hive.jdbc.url",
            "hive.jdbc.username",
            "hive.jdbc.password",
            "hive.batch.size",
            "hive.batch.timeout"
        };
        
        for (String key : hiveKeys) {
            String newValue = environment.getProperty(key);
            String oldValue = (String) configCache.get(key);
            
            if (!Objects.equals(newValue, oldValue)) {
                logger.info("检测到Hive配置变更: {} = {} -> {}", key, oldValue, newValue);
                configCache.put(key, newValue);
                notifyConfigChange(key, oldValue, newValue);
            }
        }
    }

    /**
     * 检查Groovy配置变更
     */
    private void checkGroovyConfigChanges() {
        String[] groovyKeys = {
            "groovy.script.path",
            "groovy.script.cache.enabled",
            "groovy.script.cache.size"
        };
        
        for (String key : groovyKeys) {
            String newValue = environment.getProperty(key);
            String oldValue = (String) configCache.get(key);
            
            if (!Objects.equals(newValue, oldValue)) {
                logger.info("检测到Groovy配置变更: {} = {} -> {}", key, oldValue, newValue);
                configCache.put(key, newValue);
                notifyConfigChange(key, oldValue, newValue);
            }
        }
    }

    /**
     * 检查告警配置变更
     */
    private void checkAlertConfigChanges() {
        String[] alertKeys = {
            "alert.mail.enabled",
            "alert.mail.smtp.host",
            "alert.mail.smtp.port",
            "alert.mail.recipients"
        };
        
        for (String key : alertKeys) {
            String newValue = environment.getProperty(key);
            String oldValue = (String) configCache.get(key);
            
            if (!Objects.equals(newValue, oldValue)) {
                logger.info("检测到告警配置变更: {} = {} -> {}", key, oldValue, newValue);
                configCache.put(key, newValue);
                notifyConfigChange(key, oldValue, newValue);
            }
        }
    }

    /**
     * 处理文件监听事件
     */
    private void processFileWatchEvents() {
        if (watchService == null) {
            return;
        }
        
        try {
            WatchKey key = watchService.poll();
            if (key != null) {
                for (WatchEvent<?> event : key.pollEvents()) {
                    WatchEvent.Kind<?> kind = event.kind();
                    Path context = (Path) event.context();
                    
                    if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                        logger.info("检测到配置文件修改: {}", context);
                        // 延迟处理，避免文件正在写入
                        scheduler.schedule(() -> handleConfigFileChange(context), 2, TimeUnit.SECONDS);
                    }
                }
                key.reset();
            }
        } catch (Exception e) {
            logger.error("处理文件监听事件失败", e);
        }
    }

    /**
     * 处理配置文件变更
     */
    private void handleConfigFileChange(Path configFile) {
        try {
            logger.info("处理配置文件变更: {}", configFile);
            
            // 重新加载配置
            loadInitialConfigs();
            
            // 发送配置变更事件
            eventPublisher.publishEvent(new ConfigChangeEvent("file", configFile.toString(), "modified"));
            
            // 发送告警
            alertManager.sendConfigChangeAlert("配置文件变更", configFile.toString());
            
        } catch (Exception e) {
            logger.error("处理配置文件变更失败: {}", configFile, e);
        }
    }

    /**
     * 通知配置变更
     */
    private void notifyConfigChange(String key, Object oldValue, Object newValue) {
        try {
            // 通知注册的监听器
            List<ConfigChangeListener> listeners = configListeners.get(key);
            if (listeners != null) {
                for (ConfigChangeListener listener : listeners) {
                    try {
                        listener.onConfigChange(key, oldValue, newValue);
                    } catch (Exception e) {
                        logger.error("配置变更监听器执行失败: {}", key, e);
                    }
                }
            }
            
            // 发送配置变更事件
            eventPublisher.publishEvent(new ConfigChangeEvent(key, oldValue, newValue));
            
            // 发送告警
            alertManager.sendConfigChangeAlert("配置热更新", key + " = " + newValue);
            
        } catch (Exception e) {
            logger.error("通知配置变更失败: {}", key, e);
        }
    }

    /**
     * 注册配置变更监听器
     * 
     * @param configKey 配置键
     * @param listener 监听器
     */
    public void registerConfigChangeListener(String configKey, ConfigChangeListener listener) {
        configListeners.computeIfAbsent(configKey, k -> new ArrayList<>()).add(listener);
        logger.info("注册配置变更监听器: {}", configKey);
    }

    /**
     * 取消注册配置变更监听器
     * 
     * @param configKey 配置键
     * @param listener 监听器
     */
    public void unregisterConfigChangeListener(String configKey, ConfigChangeListener listener) {
        List<ConfigChangeListener> listeners = configListeners.get(configKey);
        if (listeners != null) {
            listeners.remove(listener);
            if (listeners.isEmpty()) {
                configListeners.remove(configKey);
            }
        }
        logger.info("取消注册配置变更监听器: {}", configKey);
    }

    /**
     * 获取配置值
     * 
     * @param key 配置键
     * @return 配置值
     */
    public Object getConfig(String key) {
        return configCache.get(key);
    }

    /**
     * 获取所有配置
     * 
     * @return 配置映射
     */
    public Map<String, Object> getAllConfigs() {
        return new HashMap<>(configCache);
    }

    /**
     * 启用/禁用配置热更新
     * 
     * @param enabled 是否启用
     */
    public void setHotReloadEnabled(boolean enabled) {
        this.hotReloadEnabled = enabled;
        logger.info("配置热更新状态: {}", enabled ? "启用" : "禁用");
    }

    /**
     * 获取配置热更新状态
     * 
     * @return 是否启用
     */
    public boolean isHotReloadEnabled() {
        return hotReloadEnabled;
    }

    /**
     * 获取配置热更新统计信息
     * 
     * @return 统计信息
     */
    public ConfigHotReloadStatistics getStatistics() {
        return new ConfigHotReloadStatistics(
            configCache.size(),
            configListeners.size(),
            hotReloadEnabled,
            System.currentTimeMillis()
        );
    }

    /**
     * 销毁资源
     */
    public void destroy() {
        try {
            if (scheduler != null && !scheduler.isShutdown()) {
                scheduler.shutdown();
                scheduler.awaitTermination(5, TimeUnit.SECONDS);
            }
            
            if (watchService != null) {
                watchService.close();
            }
            
            logger.info("配置热更新管理器已销毁");
            
        } catch (Exception e) {
            logger.error("销毁配置热更新管理器失败", e);
        }
    }

    /**
     * 配置变更监听器接口
     */
    @FunctionalInterface
    public interface ConfigChangeListener {
        /**
         * 配置变更回调
         * 
         * @param key 配置键
         * @param oldValue 旧值
         * @param newValue 新值
         */
        void onConfigChange(String key, Object oldValue, Object newValue);
    }

    /**
     * 配置变更事件
     */
    public static class ConfigChangeEvent {
        private final String key;
        private final Object oldValue;
        private final Object newValue;
        private final long timestamp;

        public ConfigChangeEvent(String key, Object oldValue, Object newValue) {
            this.key = key;
            this.oldValue = oldValue;
            this.newValue = newValue;
            this.timestamp = System.currentTimeMillis();
        }

        // Getter方法
        public String getKey() { return key; }
        public Object getOldValue() { return oldValue; }
        public Object getNewValue() { return newValue; }
        public long getTimestamp() { return timestamp; }
    }

    /**
     * 配置热更新统计信息
     */
    public static class ConfigHotReloadStatistics {
        private final int totalConfigs;
        private final int totalListeners;
        private final boolean hotReloadEnabled;
        private final long timestamp;

        public ConfigHotReloadStatistics(int totalConfigs, int totalListeners, boolean hotReloadEnabled, long timestamp) {
            this.totalConfigs = totalConfigs;
            this.totalListeners = totalListeners;
            this.hotReloadEnabled = hotReloadEnabled;
            this.timestamp = timestamp;
        }

        // Getter方法
        public int getTotalConfigs() { return totalConfigs; }
        public int getTotalListeners() { return totalListeners; }
        public boolean isHotReloadEnabled() { return hotReloadEnabled; }
        public long getTimestamp() { return timestamp; }
    }
}
